package team.bluepen.supermarket.service;

import org.springframework.stereotype.Service;
import team.bluepen.supermarket.conf.KafkaProperties;
import team.bluepen.supermarket.data.database.SupermarketSystemDatabase;
import team.bluepen.supermarket.data.database.dao.ProductIdStatusAmountDao;
import team.bluepen.supermarket.data.entity.ProductIdStatusAmount;
import team.bluepen.supermarket.data.entity.ProductIdStatusPair;
import team.bluepen.supermarket.kafka.KafkaReceiveCallback;
import team.bluepen.supermarket.kafka.KafkaReceiver;
import team.bluepen.supermarket.kafka.KafkaTopics;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * @author RollW
 */
@Service
public class ProductIdStatusAmountRecvService implements KafkaReceiveCallback<ProductIdStatusPair> {
    private final Map<Long, ProductIdStatusAmount> amountMap = new HashMap<>();
    private final KafkaReceiver<ProductIdStatusPair> amountKafkaReceiver;
    private final ProductIdStatusAmountDao statusAmountDao;

    public ProductIdStatusAmountRecvService(KafkaProperties kafkaProperties,
                                            SupermarketSystemDatabase database) {
        amountKafkaReceiver = new KafkaReceiver<>(
                KafkaTopics.SQL_STATE_DATA_TOPIC,
                kafkaProperties.getBootstrapServers(),
                "ProductIdStatusAmountRecvServ",
                null,
                ProductIdStatusPair.deserializer(),
                this);
        amountKafkaReceiver.start();
        statusAmountDao = database.getProductIdStatusAmountDao();
    }

    public Collection<ProductIdStatusAmount> getAmounts() {
        return amountMap.values();
    }

    @Override
    public void onReceive(ProductIdStatusPair data) {
        ProductIdStatusAmount old = amountMap.get(data.getId());
        if (old != null) {
            ProductIdStatusAmount newOne = ProductIdStatusAmount.plus(old, data);
            amountMap.put(data.getId(), newOne);
            statusAmountDao.update(newOne);
            return;
        }
        ProductIdStatusAmount newOne = ProductIdStatusAmount.of(data);
        amountMap.put(data.getId(), newOne);
        statusAmountDao.insert(newOne);
    }
}
